package com.ibs.hadoop.hbase.dao.base;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;

import com.ibs.hadoop.hbase.bean.HbaseDataBean;
import com.ibs.hadoop.hbase.console.PropertiesEnumKey;
import com.ibs.hadoop.hbase.util.DbUtils;
import com.ibs.hadoop.hbase.util.PropertiesUtils;

/**
 * 进行hbase 数据库查询操作
* 源文件名：HbaseProcDaoImpl.java
* 文件版本：1.0.0
* 创建作者：liujun
* 创建日期：2016年10月27日
* 修改作者：liujun
* 修改日期：2016年10月27日
* 文件描述：liujun
* 版权所有：Copyright 2016 zjhz, Inc. All Rights Reserved.
*/
public class HbaseProceDaoBase {

    /**
     * 行结尾标识，用于在进行分页时排除当前行
     */
    final byte[] POSTFIX = new byte[] { 0x00 };

    /**
     * 远程连接配制信息
     */
    private final Configuration conf;

    public HbaseProceDaoBase() {
        conf = HBaseConfiguration.create();

        // 加载配制
        //conf.addResource(HbaseProceDaoBase.class.getClassLoader().getResourceAsStream("hbasecfg/hbase-site.xml"));

        conf.set("hbase.zookeeper.property.clientPort",
                PropertiesUtils.getInstance().getValue(PropertiesEnumKey.ZOOKEEPER_CLIENT_PORT.getKey()));
        conf.set("hbase.zookeeper.quorum",
                PropertiesUtils.getInstance().getValue(PropertiesEnumKey.ZOOKEEPER_CLIENT_IPADDR.getKey()));
        // conf.set("hbase.master",
        // PropertiesUtils.getInstance().getValue(PropertiesEnumKey.HBASE_MASTER_IPADDR.getKey()));
    }

    /**
     * 获得连接信息
    * 方法描述
    * @创建日期 2016年10月27日
    */
    protected Connection getConn() {
        Connection hbaseConn = null;
        // 得到连接信息
        try {
            hbaseConn = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }

        return hbaseConn;
    }

    /**
     * 进行查询操作,按行键提取10000以内在记录
     * 
     * @param table
     * @return
     */
    protected List<HbaseDataBean> query(String tableName, String rowKey) throws Exception {

        List<HbaseDataBean> list = new ArrayList<HbaseDataBean>();

        HTable table = null;
        Connection hbaseConn = null;
        try {

            hbaseConn = this.getConn();

            table = (HTable) hbaseConn.getTable(TableName.valueOf(tableName));
            Get key = new Get(rowKey.getBytes());
            // 默认过滤10000条以内的记录
            key.setFilter(new PageFilter(100000));

            Result rs = table.get(key);

            for (Cell cell : rs.listCells()) {
                list.add(getLineQueryData(cell));
            }
        } catch (IOException e) {
            throw e;
        } finally {
            DbUtils.cloaseDb(table);
            DbUtils.cloaseDb(hbaseConn);
        }

        return list;
    }

    /**
     * 进行查询操作,按行键提取num条记录
     * 
     * @param table
     * @return
     */
    protected List<HbaseDataBean> query(String tableName, String rowKey, long num) throws Exception {

        List<HbaseDataBean> list = new ArrayList<HbaseDataBean>();

        HTable table = null;
        Connection hbaseConn = null;
        try {

            hbaseConn = this.getConn();

            table = (HTable) hbaseConn.getTable(TableName.valueOf(tableName));
            Get key = new Get(rowKey.getBytes());
            // 过滤行数
            key.setFilter(new PageFilter(num));
            Result rs = table.get(key);

            for (Cell cell : rs.listCells()) {
                list.add(getLineQueryData(cell));
            }
        } catch (IOException e) {
            throw e;
        } finally {
            DbUtils.cloaseDb(table);
            DbUtils.cloaseDb(hbaseConn);
        }

        return list;
    }

    /**
     * 获取行记录数据
    * 方法描述
    * @param cell
    * @return
    * @创建日期 2016年10月27日
    */
    private HbaseDataBean getLineQueryData(Cell cell) {

        HbaseDataBean bean = new HbaseDataBean();
        // 行键
        bean.setKey(new String(CellUtil.cloneRow(cell)));
        // 列族
        bean.setColumnFamily(new String(CellUtil.cloneFamily(cell)));
        // 列
        bean.setColumnQualifier(new String(CellUtil.cloneQualifier(cell)));
        // 时间戳
        bean.setTimestamp(cell.getTimestamp());
        // 值
        bean.setValue(new String(CellUtil.cloneValue(cell)));

        return bean;

    }

    /**
     * 进行查询操作,提取一个表的前rownum记录
     * 
     * @param table 表名 
     * @param rownum 行数
     * @return
     */
    protected List<HbaseDataBean> queryByNum(String tableName, int rowNum) throws Exception {

        List<HbaseDataBean> resultList = new ArrayList<HbaseDataBean>();

        HTable table = null;
        ResultScanner rs = null;
        Connection hbaseConn = null;
        try {

            hbaseConn = this.getConn();
            table = (HTable) hbaseConn.getTable(TableName.valueOf(tableName));

            Scan scan = new Scan();

            // 过滤掉的行数
            scan.setFilter(new PageFilter(rowNum));

            scan.setMaxResultSize(rowNum);

            rs = table.getScanner(scan);

            HbaseDataBean bean = null;

            for (Result result : rs) {
                Cell[] kv = result.rawCells();

                for (int i = 0; i < kv.length; i++) {

                    bean = new HbaseDataBean();
                    bean.setKey(new String(CellUtil.cloneRow(kv[i])));
                    bean.setColumnFamily(new String(CellUtil.cloneFamily(kv[i])));
                    bean.setColumnQualifier(new String(CellUtil.cloneFamily(kv[i])));
                    bean.setTimestamp(kv[i].getTimestamp());
                    bean.setValue(new String(CellUtil.cloneValue(kv[i])));

                    resultList.add(bean);
                }
            }

        } catch (IOException e) {
            throw e;
        } finally {
            DbUtils.cloaseDb(rs);
            DbUtils.cloaseDb(table);
            DbUtils.cloaseDb(hbaseConn);
        }

        return resultList;
    }

    /**
     * 进行分页查询操作
     * 
     * @param tableName 表名
     * @param rowNum 条数
     * @param startRow 开始行
     * @param stopRow 结束行
     * @param lastRowBytes 上一次查询的尾行
     * @return
     */
    protected List<HbaseDataBean> queryPage(String tableName, int rowNum, String startRow, String stopRow,
            String lastRowBytes) throws Exception {
        // 行结束
        HTable table = null;
        ResultScanner rs = null;
        List<HbaseDataBean> list = null;

        Connection hbaseConn = null;
        try {

            hbaseConn = this.getConn();

            table = (HTable) hbaseConn.getTable(TableName.valueOf(tableName));

            Scan scan = new Scan();

            // 满足所有条件
            FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
            // 设置过滤的行数
            Filter filterPage = new PageFilter(rowNum);
            filterList.addFilter(filterPage);

            // 设置过滤信息
            scan.setFilter(filterList);
            scan.setMaxResultSize(rowNum);

            // 设置开始行号
            if (null != lastRowBytes) {
                scan.setStartRow(Bytes.add(lastRowBytes.getBytes(), POSTFIX));
            } else {
                scan.setStartRow(Bytes.toBytes(startRow));
            }

            // 设置结束号行
            if (!StringUtils.isEmpty(stopRow)) {
                scan.setStopRow(Bytes.toBytes(stopRow));
            }

            rs = table.getScanner(scan);

            list = new ArrayList<HbaseDataBean>(rowNum);

            for (Result result : rs) {
                Cell[] kv = result.rawCells();
                for (int i = 0; i < kv.length; i++) {
                    list.add(this.getLineQueryData(kv[i]));
                }
            }

        } catch (IOException e) {
            throw e;
        } finally {
            DbUtils.cloaseDb(rs);
            DbUtils.cloaseDb(table);
            DbUtils.cloaseDb(hbaseConn);
        }

        return list;
    }

    /**
     * 进行默认的查询操作,提取所有记录
     * 
     * @param tableName 表名
     * @param rowNum 条数
     * @param startRow 开始行
     * @param stopRow 结束行
     * @param lastRowBytes 上一次查询的尾行
     * @return
     */
    protected List<HbaseDataBean> queryDef(String tableName, String startRow, String stopRow) throws Exception {
        // 行结束
        HTable table = null;
        ResultScanner rs = null;
        List<HbaseDataBean> list = null;


        Connection hbaseConn = null;
        try {

            hbaseConn = this.getConn();

            table = (HTable) hbaseConn.getTable(TableName.valueOf(tableName));

            Scan scan = new Scan();


            // 设置开始行号
            scan.setStartRow(Bytes.toBytes(startRow));

            // 设置结束号行
            if (!StringUtils.isEmpty(stopRow)) {
                scan.setStopRow(Bytes.toBytes(stopRow));
            }

            rs = table.getScanner(scan);

            list = new ArrayList<HbaseDataBean>();

            for (Result result : rs) {
                Cell[] kv = result.rawCells();
                for (int i = 0; i < kv.length; i++) {
                    list.add(this.getLineQueryData(kv[i]));
                }
            }
        } finally {
            DbUtils.cloaseDb(rs);
            DbUtils.cloaseDb(table);
            DbUtils.cloaseDb(hbaseConn);
        }

        return list;
    }

    public static void main(String[] args) throws Exception {
        // HbaseProceDaoImpl base = new HbaseProceDaoImpl();
        //
        // List<HbaseDataBean> list = base.query("student", "1");
        //
        // System.out.println(list);
        //
        // System.out.println();
        //
        // List<HbaseDataBean> list2 = base.queryByNum("student", 10);
        //
        // System.out.println(list2);

        HbaseProceDaoBase daoBase = new HbaseProceDaoBase();
        List<HbaseDataBean> list = daoBase.query("H4GET_161102", "00006617122006420160508082330000847");
        
        
        System.out.println("list:"+list);
    }

}
